package com.ysy.mq;
import java.util.Objects;


import com.alibaba.fastjson2.JSON;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.ysy.constant.RocketMqConstant;
;
import com.ysy.entity.NoteLike;
import com.ysy.entity.dto.NoteLikeOrUnlikeDTO;
import com.ysy.mapper.NoteLikeMapper;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(consumerGroup = "persimmons_group_"+ RocketMqConstant.TOPIC_DELAY_DEL_NOTE_REDIS_CACHE,
        topic = RocketMqConstant.TOPIC_DELAY_DEL_NOTE_REDIS_CACHE,
        consumeMode = ConsumeMode.ORDERLY)
public class NoteLikeOrUnLikeConsumer implements RocketMQListener<Message> {

    RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private NoteLikeMapper noteLikeMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        rateLimiter.acquire();
        String body = new String(message.getBody());
        NoteTagMessageHandle(body);
    }

    private void NoteTagMessageHandle(String body) {
        NoteLikeOrUnlikeDTO noteLikeOrUnlikeDTO = JSON.parseObject(body, NoteLikeOrUnlikeDTO.class);
        if (Objects.isNull(noteLikeOrUnlikeDTO)) return;
        // 执行落库
        NoteLike noteLike = new NoteLike();
        noteLike.setUserId(noteLikeOrUnlikeDTO.getUserId());
        noteLike.setNoteId(noteLikeOrUnlikeDTO.getNoteId());
        noteLike.setCreateTime(noteLikeOrUnlikeDTO.getCreateTime());
        noteLike.setStatus(noteLikeOrUnlikeDTO.getType());
      //  boolean result = noteLikeMapper.insertOrUpdate(noteLike); 效率低查询2次
        int count = noteLikeMapper.insertOrUpdateSql(noteLike);
        if (count == 0) return;
        // 落库成功进行mq计数
        MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(body);
        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(RocketMqConstant.NOTE_LIKE_OR_UNLIKE_COUNT_TOPIC, messageBuilder, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
            }

            @Override
            public void onException(Throwable throwable) {
            }
        });
    }
}
